package com.example.demo.spark;

import java.util.ArrayList;
import java.util.List;

import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;
import org.apache.spark.sql.functions;
import org.apache.spark.sql.types.DataTypes;
import org.apache.spark.sql.types.StructField;
import org.apache.spark.sql.types.StructType;
@SuppressWarnings("all")
public class SparkCsv2Parquet {
    public static void main(String[] args) {
        if (args.length < 2) {
            System.err.println("Usage: SparkCsv2Parquet <csv dir> <parquet dir>");
            System.exit(1);
        }
        String srcPath = args[0];
        String targetPath = args[1];

        SparkSession spark = SparkSession.builder()
            .appName("SparkCsv2Parquet")
            .getOrCreate();

        //Dataset<Row> df = spark.read().schema(createSchema()).csv(srcPath);
        Dataset<Row> df = spark.read().parquet(srcPath);
        df = df.select("name","mobile","addr","tradename");
        df.write().parquet(targetPath);

        df.where(functions.col("addr"));

        spark.stop();
    }

    private static StructType createSchema() {
        String schemaString = "dbname,ac0018,name,mobile,tel,addr,org_addr,freight,totalcommodityvalue," +
            "totalorderamount,paytime,buyerid,platform,email,refundamount,discountamount," +
            "ordersubmissiontime,shippedqty,univalent,unitDiscountAmount,barcode,sku," +
            "tradename,tradename_category,brand,tracking_number,express_company";

        List<StructField> fields = new ArrayList<>();
        for (String fieldName : schemaString.split(",")) {
            StructField field = DataTypes.createStructField(fieldName, DataTypes.StringType, true);
            fields.add(field);
        }

        StructType schema = DataTypes.createStructType(fields);
        return schema;
    }

}
